package com.sojson.util.mq.rabbit.code.interfaces.impl;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import com.rabbitmq.client.AMQP.BasicProperties.Builder;
import com.rabbitmq.client.Channel;
import com.sojson.util.StringUtil;
import com.sojson.util.mq.rabbit.enums.ERabbitExchange;

/**
 * rabbitmq_delayed_message_exchange插件</BR>
 * 用来让消息延迟一段时间再发送到交换机,以实现延迟发送消息</BR>
 * 创建交换机的时候需要在参数里面指定交换机的类型{@link com.sojson.util.mq.rabbit.enums.ERabbitExchange}</BR>
 * 发送的消息如果要实现延迟发送,需要在发送的参数里面指定延迟时间(单位:毫秒)</BR>
 * 
 * @author liu
 * @date 2022-06-16
 */
public class RabbitCodeDelayMsgUtilImpl {

    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeDirectUtilImpl}的扩展类 */
    private static RabbitCodeDirectDelayMsgUtilImpl direct;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeTopicUtilImpl}的扩展类 */
    private static RabbitCodeTopicDelayMsgUtilImpl topic;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeFanoutUtilImpl}的扩展类 */
    private static RabbitCodeFanoutDelayMsgUtilImpl fanout;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeHeadersUtilImpl}的扩展类 */
    private static RabbitCodeHeadersDelayMsgUtilImpl headers;

    /** 获取 {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeDirectUtilImpl}的扩展类 */
    public static RabbitCodeDirectDelayMsgUtilImpl getDirect() {
        if (isBlankObject(direct)) {
            direct = new RabbitCodeDelayMsgUtilImpl().new RabbitCodeDirectDelayMsgUtilImpl();
        }
        return direct;
    }

    /** 获取 {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeTopicUtilImpl}的扩展类 */
    public static RabbitCodeTopicDelayMsgUtilImpl getTopic() {
        if (isBlankObject(topic)) {
            topic = new RabbitCodeDelayMsgUtilImpl().new RabbitCodeTopicDelayMsgUtilImpl();
        }
        return topic;
    }

    /** 获取 {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeFanoutUtilImpl}的扩展类 */
    public static RabbitCodeFanoutDelayMsgUtilImpl getFanout() {
        if (isBlankObject(fanout)) {
            fanout = new RabbitCodeDelayMsgUtilImpl().new RabbitCodeFanoutDelayMsgUtilImpl();
        }
        return fanout;
    }

    /** 获取{@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeHeadersUtilImpl}的扩展类 */
    public static RabbitCodeHeadersDelayMsgUtilImpl getHeaders() {
        if (isBlankObject(headers)) {
            headers = new RabbitCodeDelayMsgUtilImpl().new RabbitCodeHeadersDelayMsgUtilImpl();
        }
        return headers;
    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeDirectUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitCodeDirectDelayMsgUtilImpl extends RabbitCodeDirectUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String routingKey, String body, Long delayTime)
            throws IOException {
            basicPublish(channel, exchange, routingKey, false, false, body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String routingKey, boolean mandatory,
            boolean immediate, String body, Long delayTime) throws IOException {
            basicPublish(channel, exchange, routingKey, mandatory, immediate, getBasicProperties(), body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String routingKey, boolean mandatory,
            boolean immediate, Builder props, String body, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            Map<String, Object> headers = props.build().getHeaders();
            if (isBlankObject(headers)) {
                headers = new HashMap<>(10);
            }
            if (isNotBlankObject(delayTime)) {
                Map<String, Object> consumeHeaderParam = getConsumeHeaderParam(delayTime);
                headers.putAll(consumeHeaderParam);
            }
            props = props.headers(headers);
            channel.basicPublish(getExchangeName(exchange), getRoutingKeyName(routingKey),
                mandatory, immediate, props.build(), body.getBytes());
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.DIRECT);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitCodeDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeTopicUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitCodeTopicDelayMsgUtilImpl extends RabbitCodeTopicUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String routingKey, String body, Long delayTime)
            throws IOException {
            basicPublish(channel, exchange, routingKey, false, body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String routingKey, boolean mandatory, String body,
            Long delayTime) throws IOException {
            basicPublish(channel, exchange, routingKey, mandatory, false, getBasicProperties(), body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String routingKey, boolean mandatory,
            boolean immediate, Builder props, String body, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            Map<String, Object> headers = props.build().getHeaders();
            if (isBlankObject(headers)) {
                headers = new HashMap<>(10);
            }
            if (isNotBlankObject(delayTime)) {
                Map<String, Object> consumeHeaderParam = getConsumeHeaderParam(delayTime);
                headers.putAll(consumeHeaderParam);
            }
            props = props.headers(headers);
            channel.basicPublish(getExchangeName(exchange), getRoutingKeyName(routingKey),
                mandatory, immediate, props.build(), body.getBytes());
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.TOPIC);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitCodeDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeFanoutUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitCodeFanoutDelayMsgUtilImpl extends RabbitCodeFanoutUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String body, Long delayTime) throws IOException {
            basicPublish(channel, exchange, false, false, body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, boolean mandatory, boolean immediate, String body,
            Long delayTime) throws IOException {
            basicPublish(channel, exchange, mandatory, immediate, getBasicProperties(), body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, boolean mandatory, boolean immediate, Builder props,
            String body, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            Map<String, Object> headers = props.build().getHeaders();
            if (isBlankObject(headers)) {
                headers = new HashMap<>(10);
            }
            if (isNotBlankObject(delayTime)) {
                Map<String, Object> consumeHeaderParam = getConsumeHeaderParam(delayTime);
                headers.putAll(consumeHeaderParam);
            }
            props = props.headers(headers);
            channel.basicPublish(getExchangeName(exchange), StringUtil.EMPTY, mandatory, immediate,
                props.build(), body.getBytes());
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.FANOUT);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitCodeDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeHeadersUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitCodeHeadersDelayMsgUtilImpl extends RabbitCodeHeadersUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param headerParam   请求头参数
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, String body, Map<String, Object> headerParam,
            Long delayTime) throws IOException {
            basicPublish(channel, exchange, false, false, body, headerParam, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param body          内容
         * @param headerParam   请求头参数
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, boolean mandatory, boolean immediate, String body,
            Map<String, Object> headerParam, Long delayTime) throws IOException {
            basicPublish(channel, exchange, mandatory, immediate, getBasicProperties(), body, headerParam, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param headerParam   请求头参数
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(Channel channel, String exchange, boolean mandatory, boolean immediate, Builder props,
            String body, Map<String, Object> headerParam, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            Map<String, Object> headers = props.build().getHeaders();
            if (isBlankObject(headers)) {
                headers = new HashMap<>(10);
            }
            if (isNotBlankObject(headerParam)) {
                headers.putAll(headerParam);
            }
            if (isNotBlankObject(delayTime)) {
                Map<String, Object> consumeHeaderParam = getConsumeHeaderParam(delayTime);
                headers.putAll(consumeHeaderParam);
            }
            props = props.headers(headers);
            channel.basicPublish(getExchangeName(exchange), StringUtil.EMPTY, mandatory, immediate,
                props.build(), body.getBytes());
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.HEADERS);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitCodeDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * 初始化交换机参数
     * 
     * @param param
     */
    public static void initExchangeParam(Map<String, Object> param, ERabbitExchange type) {
        param.put("x-delayed-type", type.getCode());
    }

    /**
     * 获取交换机类型
     * 
     * @return
     */
    protected static String getExchangeType() {
        return ERabbitExchange.DELAYED_MESSAGE.getCode();
    }

    /**
     * 获取消息头参数
     * 
     * @param delayTime 发送延迟时间
     * @return
     */
    protected Map<String, Object> getConsumeHeaderParam(Long delayTime) {
        Map<String, Object> headers = new HashMap<>(1);
        // x-delay过期时间(单位: 毫秒),需要配合rabbitmq_delayed_message_exchange插件使用
        headers.put("x-delay", delayTime);
        return headers;
    }

    /**
     * 判断一个对象是否是空(空对象)
     * 
     * @param obj
     * @return
     */
    protected static boolean isBlankObject(Object obj) {
        return StringUtil.isBlankObject(obj);
    }

    /**
     * 判断一个对象是否不是空(空对象)
     * 
     * @param obj
     * @return
     */
    protected boolean isNotBlankObject(Object obj) {
        return StringUtil.isNotBlankObject(obj);
    }

}